package org.example.data_stream;

import cn.hutool.core.util.RandomUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.concurrent.TimeUnit;

/**
 * @author shenguangyang
 */
public class E12_SinkToRedis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Event> streamSource = env.addSource(new ClickSource());

        // 将结果数据写入到redis中
        FlinkJedisPoolConfig poolConfig = new FlinkJedisPoolConfig.Builder()
                .setPassword("project_dev@nwt")
                .setPort(56379)
                .setTimeout(2000)
                .setHost("work01")
                .build();

        // 自定义类实现 RedisMapper 接口
        streamSource.addSink(new RedisSink<>(poolConfig, new MyRedisMapper()));

        env.execute();
    }

    /**
     * 用户点击源
     * @author shenguangyang

     */
    private static class ClickSource implements SourceFunction<Event> {
        // 声明一个标志位
        private volatile Boolean cancel = false;

        @Override
        public void run(SourceContext<Event> ctx) throws Exception {
            String[] urls = {"./home", "./cart", "./fav", "./prod?id=100", "./prod?id=10"};
            String[] users = {"Mary", "Alice", "Bod", "Cary"};
            while (!cancel) {
                ctx.collect(new Event(users[RandomUtil.randomInt(0, users.length)], urls[RandomUtil.randomInt(0, urls.length)], System.currentTimeMillis()));
                TimeUnit.MILLISECONDS.sleep(200);
            }
        }

        @Override
        public void cancel() {
            this.cancel = true;
        }
    }

    private static class MyRedisMapper implements RedisMapper<Event> {
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "clicks");
        }

        @Override
        public String getKeyFromData(Event event) {
            return event.getUser();
        }

        @Override
        public String getValueFromData(Event event) {
            return event.getUrl();
        }
    }
}
